package top.v5it.logback.push.convergence;

import lombok.Getter;
import lombok.Setter;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * 滑动窗口收敛策略
 *
 * @author zhanpu
 * @date 2022/9/13
 */
@Setter
@Getter
public class SlidingWindowConvergenceStrategy implements ConvergenceStrategy {

    private final static ConcurrentMap<String, DelayQueue<MarkMessage>> delay = new ConcurrentHashMap<>();

    /**
     * 限制大小
     */
    private long limit;

    /**
     * <b>窗口大小</b>
     */
    private long windowSize;

    /**
     * 统计窗口范围内单条消息到达数量，当窗口范围内单条消息数量到达{@link SlidingWindowConvergenceStrategy#limit}限制就推送此条消息并清除窗口统计
     *
     * @param payload 消息体
     * @return
     */
    @Override
    public boolean isPush(byte[] payload) {
        return getMsgId(payload).map(t -> {
            delay.computeIfAbsent(t, k -> new DelayQueue<>());
            final DelayQueue<MarkMessage> markMessages =
                    delay.computeIfPresent(t, (k, v) -> {
                        v.offer(new MarkMessage(windowSize));
                        final int size = v.drainTo(new ArrayList<>());
                        return size >= limit ? null : v;
                    });
            return Objects.isNull(markMessages);
        }).orElse(Boolean.FALSE);
    }

    static class MarkMessage implements Delayed, Serializable {

        private static final long serialVersionUID = -3053936720788012411L;
        private final long expired;

        MarkMessage(long windowSize) {
            this.expired = ConvergenceStrategy.toEpochSecond() + windowSize;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            final long toEpochSecond = ConvergenceStrategy.toEpochSecond();
            return unit.convert(expired - toEpochSecond, TimeUnit.SECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            MarkMessage entity = (MarkMessage) o;
            return (int) (this.getDelay(TimeUnit.SECONDS) - entity.getDelay(TimeUnit.SECONDS));
        }

    }
}
